import boot.Boot;
import common.RecordListener;
import boot.MysqlRecordPrinter;
import common.UserRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.Map;
import java.util.Properties;

import recordgenerator.Names;

public class NotifyDemo {
    private static final Logger log = LoggerFactory.getLogger(NotifyDemo.class);

    public static Map<String, RecordListener> buildRecordListener() {
        // user can impl their own listener
        RecordListener mysqlRecordPrintListener = new RecordListener() {
            @Override
            public void consume(UserRecord record) {
                // consume record
                // MysqlRecordPrinter show how to go through record fields and get general attributes
                String ret = MysqlRecordPrinter.recordToString(record.getRecord());

                log.info(ret);
                record.commit(String.valueOf(record.getRecord().getSourceTimestamp()));
            }
        };
        return Collections.singletonMap("mysqlRecordPrinter", mysqlRecordPrintListener);
    }

    /**
     * This demo use  hard coded config. User can modify variable value for test
     * The detailed describe for var in resources/demoConfig
     */
    public static Properties getConfigs() {
        Properties properties = new Properties();
        // user password and sid for auth                "");
        properties.setProperty(Names.USER_NAME, "liwy3");
        properties.setProperty(Names.PASSWORD_NAME, "liweiye2j");
        properties.setProperty(Names.SID_NAME, "dtss5f11xyg16jmbc1");
        // kafka consumer group general same with sid
        properties.setProperty(Names.GROUP_NAME, "test12");
        // topic to consume, partition is 0
        properties.setProperty(Names.KAFKA_TOPIC, "cn_hangzhou_47.95.150.62_3307_root_version2");
        // kafka broker url
        properties.setProperty(Names.KAFKA_BROKER_URL_NAME, "dts-cn-hangzhou.aliyuncs.com:18001");
        // initial checkpoint for first seek(a timestamp to set, eg 1566180200 if you want (Mon Aug 19 10:03:21 CST 2019))
        properties.setProperty(Names.INITIAL_CHECKPOINT_NAME, "1637061634");
        // if force use config checkpoint when start. for checkpoint reset
        properties.setProperty(Names.USE_CONFIG_CHECKPOINT_NAME, "true");
        // use consumer assign or subscribe interface
        // when use subscribe mode, group config is required. kafka consumer group is enabled
        properties.setProperty(Names.SUBSCRIBE_MODE_NAME, "assign");
        return properties;
    }

    public static void main(String[] args) throws InterruptedException {

        try {
            Boot.boot(getConfigs(), buildRecordListener());
        } catch (Throwable e) {
            log.error("NotifyDemo: failed cause " + e.getMessage(), e);
            throw e;
        } finally {
            System.exit(0);
        }
    }
}
